namespace Northwestern.SSIS.Components
{
    using System;
    using System.Collections;
    using System.Collections.Generic;
    using System.Data;
    using Microsoft.SqlServer.Dts.Pipeline;
    using Microsoft.SqlServer.Dts.Pipeline.Wrapper;
    using Microsoft.SqlServer.Dts.Runtime;
    using Microsoft.SqlServer.Dts.Runtime.Wrapper;
    using System.IO;
    using System.Text;
    using System.Globalization;
    using System.Windows.Forms;
    using System.Runtime.InteropServices;

    /// <summary>
    /// Asynchronous Transform that aggregates text from multiple rows.
    /// Key columns are defined and for each grouping of key columns, text
    /// from a specified column is concatenated and output to a new column.
    /// 
    /// There's some good information/code on pseudo-synchronous-asynch transforms here:
    /// http://msdn.microsoft.com/en-us/library/ms135931.aspx (Developing a Custom Transformation Component with Asynchronous Outputs)
    /// </summary>
    [DtsPipelineComponent(
        DisplayName = "AggregateText",
        Description = "After defining group by keys, aggregates text field into single row.",
        IconResource = "Northwestern.SSIS.Components.AggregateText.ico",
        ComponentType = ComponentType.Transform)]


    public class AggregateText : PipelineComponent
    {
        #region Members
        bool cancelEvent;

        int aggColumnIn_Index;
        int aggColumnOut_Index;
        List<int> keyColumn_Indices;

        // Used to store state of aggregation
        bool rowMatches;
        bool firstRow;

        string newlinechar;

        System.Text.ASCIIEncoding enc = new System.Text.ASCIIEncoding();

        // Map the Input/Output columns by hand, since this is an Async
        Dictionary<int, int> columnMapping;

        /// <summary>
        /// Runtime member. Used to cache the output buffer received in PrimeOutput. This buffer contains
        /// the rows in the output column collection of the meta data.
        /// </summary>
        PipelineBuffer outputBuffer;


        #endregion

        #region Design Time

        #region ProvideComponentProperties
        /// <summary>
        /// Called when the component is initally added to a data flow task. 
        /// Create and configure the input and outputs of the component.
        /// </summary>
        public override void ProvideComponentProperties()
        {
            /// Support resettability.
            /// The base class calls RemoveAllInputsOutputsAndCustomProperties to reset the
            /// component. Used here to highlight the functionality of the base class.
            base.RemoveAllInputsOutputsAndCustomProperties();

            // Let the base component add the input and output.
            base.ProvideComponentProperties();

            ComponentMetaData.UsesDispositions = true;

            // Name the input and output, and make the output asynchronous.
            ComponentMetaData.InputCollection[0].Name = "Input";
            ComponentMetaData.InputCollection[0].ErrorRowDisposition = DTSRowDisposition.RD_FailComponent;
            ComponentMetaData.InputCollection[0].TruncationRowDisposition = DTSRowDisposition.RD_FailComponent;

            IDTSOutput100 errorOutput = ComponentMetaData.OutputCollection.New();
            errorOutput.IsErrorOut = true;
            errorOutput.Name = "ErrorOutput";

            /// Add the output.
            IDTSOutput100 output = ComponentMetaData.OutputCollection[0];
            output.Name = "Outputs";
            output.Description = "Aggregate rows are directed to this output.";
            output.SynchronousInputID = 0;
            output.ErrorRowDisposition = DTSRowDisposition.RD_FailComponent;
            output.TruncationRowDisposition = DTSRowDisposition.RD_FailComponent;

            IDTSCustomProperty100 keyColumnsProp = ComponentMetaData.CustomPropertyCollection.New();
            keyColumnsProp.Name = "keyColumns";
            keyColumnsProp.Description = "A comma separated list of columns that uniquely identify a group of text rows to be aggregated";

            IDTSCustomProperty100 AggregateColumnProp = ComponentMetaData.CustomPropertyCollection.New();
            AggregateColumnProp.Name = "AggregateColumn";
            AggregateColumnProp.Description = "The name of the column containing text to aggregate";

            IDTSCustomProperty100 OutputColumnProp = ComponentMetaData.CustomPropertyCollection.New();
            OutputColumnProp.Name = "OutputColumn";
            OutputColumnProp.Description = "The name of the column where the aggregated text will be output";
            
            IDTSCustomProperty100 lineBreakProp = ComponentMetaData.CustomPropertyCollection.New();
            lineBreakProp.Name = "LineBreakStyle";
            lineBreakProp.Description = "The style of line break to use: (Windows or Unix). The default is Windows.";
            lineBreakProp.Value = "Windows";
        }
        #endregion

        #region Validate
        /// <summary>
        /// Called repeatedly when the component is edited in the designer, and once at the beginning of execution.
        /// Verifies the following:
        /// 1. The component has 2 outputs.
        /// 2. The component has 1 input.
        /// 3. There is an output column in each output for each input column.
        /// 4. The base class validation succeeds.
        /// </summary>
        /// <returns></returns>
        [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Globalization", "CA1303:DoNotPassLiteralsAsLocalizedParameters", MessageId = "Microsoft.Samples.SqlServer.Dts.AggregateText.InternalFireError(System.String)"), System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Design", "CA1031:DoNotCatchGeneralExceptionTypes"), CLSCompliant(false)]
        public override DTSValidationStatus Validate()
        {
            /*
            if (ComponentMetaData.OutputCollection.Count != 2)
            {
                InternalFireError("There should be two outputs. The metadata of this component is corrupt.");
                return DTSValidationStatus.VS_ISCORRUPT;
            }
            if (ComponentMetaData.InputCollection.Count != 1)
            {
                InternalFireError("There is more than one input. The metadata of this component is corrupt.");
                return DTSValidationStatus.VS_ISCORRUPT;
            }
            */
            IDTSInput100 input = ComponentMetaData.InputCollection[0];
            IDTSOutput100 output = ComponentMetaData.OutputCollection[0];

            /// If there isn't a matching output column for each input column 
            /// then the component meta data is out of sync. Return VS_NEEDSNEWMETADATA
            /// because ReinitializeMetaData method fixes this problem.
            /// 



            if (ComponentMetaData.CustomPropertyCollection["LineBreakStyle"].Value.ToString() != "Windows" &&
                ComponentMetaData.CustomPropertyCollection["LineBreakStyle"].Value.ToString() != "Unix")
            {
                InternalFireError("LineBreakStyle property must be set to either 'Windows' or 'Unix'.");
                return DTSValidationStatus.VS_ISBROKEN;
            }
            if (ComponentMetaData.CustomPropertyCollection["AggregateColumn"].Value == null)
            {
                InternalFireError("AggregateColumn property must be set.");
                return DTSValidationStatus.VS_ISBROKEN;
            }
            if (ComponentMetaData.CustomPropertyCollection["OutputColumn"].Value == null)
            {
                InternalFireError("OutputColumn property must be set.");
                return DTSValidationStatus.VS_ISBROKEN;
            }
            if (ComponentMetaData.CustomPropertyCollection["keyColumns"].Value == null)
            {
                InternalFireError("keyColumns property must be set.");
                return DTSValidationStatus.VS_ISBROKEN;
            }

            //
            // parse they keyColumns from the Custom Properties.  It is entered as a comma separated list.
            //
                // If all of those exist, then make sure they exist in the 
                // input and outputs
                if (!InputColumnExistsByName( ComponentMetaData.InputCollection[0], GetCustomPropertyString("AggregateColumn") ))
                {
                    InternalFireError("AggregateColumn: '" +
                        ComponentMetaData.CustomPropertyCollection["AggregateColumn"].Value.ToString() +
                        "' must be one of the input columns but is not found in the input");
                    return DTSValidationStatus.VS_ISBROKEN;
                }
                if (!OutputColumnExistsByName(ComponentMetaData.OutputCollection[0], GetCustomPropertyString("OutputColumn")))
                {
                    InternalFireError("OutputColumn: '" +
                        ComponentMetaData.CustomPropertyCollection["OutputColumn"].Value.ToString() +
                        "' should show up in the output, but does not.  Please contact the developer of this component.");
                    return DTSValidationStatus.VS_ISBROKEN;
                }
                foreach (string ColName in GetCustomPropertyString("keyColumns").Split(','))
                {
                    if (!InputColumnExistsByName(ComponentMetaData.InputCollection[0], ColName))
                    {
                        InternalFireError("One or more of the keyColumns: '" +
                            ColName +
                            "' could not be found in the input columns.");
                        return DTSValidationStatus.VS_ISBROKEN;
                    }


            }

            foreach (IDTSInputColumn100 col in input.InputColumnCollection)
            {
                if (col.Name != GetCustomPropertyString("AggregateColumn") && !OutputColumnExists(output, col.LineageID))
                {
                    ComponentMetaData.FireError(0, ComponentMetaData.Name, "The input column " + col.Name + " is not found in the " + output.Name + " output.", "", 0, out cancelEvent);
                    return DTSValidationStatus.VS_NEEDSNEWMETADATA;
                }

            }
            /// Finally, call the base class, which validates that the LineageID of each column in the input collection
            /// matches the LineageID of a column in the VirtualInputColumnCollection. 
            return base.Validate();
        }
        #endregion

        #region SetUsageType
        /// <summary>
        /// The SetUsageType method is called when a column from the virtual input collection is selected for use by the component.
        /// The virtual input collection contains the collection of columns that the upstream component is providing to this component.
        /// If the column DataType is DT_IMAGE it is rejected, since BLOB datatypes are not supported by this component.
        /// Based on the usageType parameter the column is either added, removed, or rejected by the component. 
        /// UsageType.UT_IGNORED - The column was previously added to the input collection. The base class removes the column from the input.
        /// the matching output columns are removed.
        /// UsageType.UT_READONLY - The column is added to the input collection of the component.
        /// UsageType.UT_READWRITE - Since this component does not modify columns this usagetype is rejected.
        /// </summary>
        /// <param name="inputID">The ID of the input the column is mapped to.</param>
        /// <param name="virtualInput">The IDTSVirtualInput100 for the input.</param>
        /// <param name="lineageID">The lineageID of the virtual input column.</param>
        /// <param name="usageType">The usagetype of the column.</param>
        /// <returns>The newly created input column.</returns>
        [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Usage", "CA2201:DoNotRaiseReservedExceptionTypes"), System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Globalization", "CA1303:DoNotPassLiteralsAsLocalizedParameters", MessageId = "System.Exception.#ctor(System.String)"), CLSCompliant(false)]
        public override IDTSInputColumn100 SetUsageType(int inputID, IDTSVirtualInput100 virtualInput, int lineageID, DTSUsageType usageType)
        {

            if (virtualInput == null)
            {
                throw new ArgumentNullException("virtualInput");
            }

            IDTSVirtualInputColumn100 vCol = virtualInput.VirtualInputColumnCollection.GetVirtualInputColumnByLineageID(lineageID);
            IDTSInputColumn100 col = null;

            /// No support for BLOB image columns.
            if (vCol.DataType == DataType.DT_IMAGE)
                throw new Exception("Binary data types not supported.");

            /// If the usageType is UT_IGNORED, then the column is being removed.
            /// So remove it from the outputs also.
            if (usageType == DTSUsageType.UT_IGNORED)
            {
                /// If the usageType is UT_IGNORED, the base class removes the column
                /// and the returned column is null. 
                col = base.SetUsageType(inputID, virtualInput, lineageID, usageType);

                /// Remove the matching output columns
                RemoveOutputColumn(ComponentMetaData.OutputCollection[0].ID, vCol.LineageID);
            }
            else if (usageType == DTSUsageType.UT_READWRITE)
            {
                throw new Exception("Read write not supported.");
            }
            else
            {
                /// Let the base class add the input column.
                col = base.SetUsageType(inputID, virtualInput, lineageID, usageType);
                /// Add an output column to the outputs.
                if (col.Name != GetCustomPropertyString("AggregateColumn"))
                    AddOutputColumn(ComponentMetaData.OutputCollection[0].ID, col);
            }
            return col;
        }
        #endregion

        #region ReinitializeMetaData
        /// <summary>
        /// Called when VS_NEEDSNEWMETADATA is returned from Validate. 
        /// Reset all of the output columns.
        /// </summary>
        [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Design", "CA1031:DoNotCatchGeneralExceptionTypes")]
        public override void ReinitializeMetaData()
        {

            AddInputColumnsToOutput(ComponentMetaData.OutputCollection[0].ID);
            AddTextOutputColumn();

        }
        #endregion

        #region OnInputPathDetached
        /// <summary>
        /// Called when the path connecting the component to an upstream component is deleted. 
        /// Clear the columns from the output collections.
        /// </summary>
        /// <param name="inputID"></param>
        public override void OnInputPathDetached(int inputID)
        {
            base.OnInputPathDetached(inputID);

            ComponentMetaData.OutputCollection[0].OutputColumnCollection.RemoveAll();
        }
        #endregion


        #region InsertInput
        /// <summary>
        /// Called to add an input to the ComponentMetaData. This component doesn't allow adding inputs,
        /// so an exception is thrown.
        /// </summary>
        /// <param name="insertPlacement"></param>
        /// <param name="inputID"></param>
        /// <returns></returns>
        [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Usage", "CA2201:DoNotRaiseReservedExceptionTypes"), System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Globalization", "CA1303:DoNotPassLiteralsAsLocalizedParameters", MessageId = "System.Exception.#ctor(System.String)"), CLSCompliant(false)]
        public override IDTSInput100 InsertInput(DTSInsertPlacement insertPlacement, int inputID)
        {
            throw new Exception("Component " + ComponentMetaData.Name + " does not allow adding inputs.");
        }
        #endregion

        #region InsertOutput
        /// <summary>
        /// Called to add an output to the ComponentMetaData. This component doesn't allow adding outputs,
        /// so an exception is thrown.
        /// </summary>
        /// <param name="insertPlacement"></param>
        /// <param name="outputID"></param>
        /// <returns></returns>
        [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Usage", "CA2201:DoNotRaiseReservedExceptionTypes"), System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Globalization", "CA1303:DoNotPassLiteralsAsLocalizedParameters", MessageId = "System.Exception.#ctor(System.String)"), CLSCompliant(false)]
        public override IDTSOutput100 InsertOutput(DTSInsertPlacement insertPlacement, int outputID)
        {
            throw new Exception("Component " + ComponentMetaData.Name + " does not allow adding outputs.");
        }
        #endregion

        #region InsertOutputColumnAt
        /// <summary>
        /// Called to add an output to the ComponentMetaData. This component doesn't allow adding outputs,
        /// so an exception is thrown.
        /// </summary>
        /// <param name="outputID"></param>
        /// <param name="outputColumnIndex"></param>
        /// <param name="name"></param>
        /// <param name="description"></param>
        /// <returns></returns>
        [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Usage", "CA2201:DoNotRaiseReservedExceptionTypes"), System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Globalization", "CA1303:DoNotPassLiteralsAsLocalizedParameters", MessageId = "System.Exception.#ctor(System.String)"), CLSCompliant(false)]
        public override IDTSOutputColumn100 InsertOutputColumnAt(int outputID, int outputColumnIndex, string name, string description)
        {
            throw new Exception("Output columns cannot be added to " + ComponentMetaData.Name);
        }
        #endregion

        #region SetOutputColumnProperty
        /// <summary>
        /// Called when the custom property on an output column is set. 
        /// Disallow modifications of the InputColumnLineageID custom property by 
        /// throwing an exception.
        /// </summary>
        /// <param name="outputID"></param>
        /// <param name="outputColumnID"></param>
        /// <param name="propertyName"></param>
        /// <param name="propertyValue"></param>
        /// <returns></returns>
        [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Usage", "CA2201:DoNotRaiseReservedExceptionTypes"), System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Globalization", "CA1303:DoNotPassLiteralsAsLocalizedParameters", MessageId = "System.Exception.#ctor(System.String)"), CLSCompliant(false)]
        public override IDTSCustomProperty100 SetOutputColumnProperty(int outputID, int outputColumnID, string propertyName, object propertyValue)
        {
            if (propertyName == "InputColumnLineageID")
                throw new Exception("The InputColumnLineageID property is read only.");

            return base.SetOutputColumnProperty(outputID, outputColumnID, propertyName, propertyValue);
        }
        #endregion
        #region SetComponentProperty
        [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Usage", "CA2201:DoNotRaiseReservedExceptionTypes"), System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Globalization", "CA1303:DoNotPassLiteralsAsLocalizedParameters", MessageId = "System.Exception.#ctor(System.String)"), CLSCompliant(false)]
        public override IDTSCustomProperty100 SetComponentProperty(
            string propertyName,
            Object propertyValue
        )
        {
            string aggColumnIn = GetCustomPropertyString("AggregateColumn");
            string aggColumnOut = GetCustomPropertyString("OutputColumn");

            IDTSCustomProperty100 returnVal = base.SetComponentProperty(propertyName, propertyValue);

            if (propertyName == "AggregateColumn")
            {
                aggColumnIn = propertyValue.ToString().Trim();
                RemoveOutputColumnByName(ComponentMetaData.OutputCollection[0], aggColumnIn);
            }
            if (propertyName == "OutputColumn")
            {
                
                RemoveOutputColumnByName(ComponentMetaData.OutputCollection[0], aggColumnOut);
                AddTextOutputColumn();
            }

            return returnVal;
        }
        # endregion

        #region SetOutputProperty
        /// <summary>
        /// Called when a custom property of an output object is set. The only custom property added and used by this
        /// component is the IsDistinctOutput custom property. It cannot be modified so an exception is thrown.
        /// </summary>
        /// <param name="outputID">The ID of the output.</param>
        /// <param name="propertyName">The name of the property.</param>
        /// <param name="propertyValue">The value to assign to the property.</param>
        /// <returns>The custom property.</returns>
        [System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Usage", "CA2201:DoNotRaiseReservedExceptionTypes"), System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Globalization", "CA1303:DoNotPassLiteralsAsLocalizedParameters", MessageId = "System.Exception.#ctor(System.String)"), CLSCompliant(false)]
        public override IDTSCustomProperty100 SetOutputProperty(int outputID, string propertyName, object propertyValue)
        {

            if (propertyName == "IsDistinctOutput")
                throw new Exception("The IsDistinctOutput property is used internally by the component and cannot be modified.");

            return base.SetOutputProperty(outputID, propertyName, propertyValue);
        }
        #endregion

        #endregion

        #region Runtime Methods

        #region PreExecute
        /// <summary>
        /// Called prior to PrimeOutput and ProcessInput. 
        /// Creates the internal buffer and stores the buffer column indexs of the input and output columns
        /// </summary>
        public override void PreExecute()
        {
            // Initialize variables used in ProcessInput
            rowMatches = true;
            firstRow = true;

            if (GetCustomPropertyString("LineBreakStyle") == "Unix")
            {
                newlinechar = "\n";
            }
            else
            {
                newlinechar = "\r\n";
            }
            //
            // Init and fill the ColumnMapping Dictionary
            // For all the columns in the input that are also present it the ouput
            // this stores a mapping of input indices to output indices
            columnMapping = new Dictionary<int, int>();
            foreach (IDTSInputColumn100 inCol in ComponentMetaData.InputCollection[0].InputColumnCollection)
            {
                foreach (IDTSOutputColumn100 outCol in ComponentMetaData.OutputCollection[0].OutputColumnCollection)
                {
                    if (inCol.Name.Equals(outCol.Name))
                    {

                        int inIndex = this.BufferManager.FindColumnByLineageID(ComponentMetaData.InputCollection[0].Buffer, inCol.LineageID);
                        int outIndex = this.BufferManager.FindColumnByLineageID(ComponentMetaData.OutputCollection[0].Buffer, outCol.LineageID);
                        columnMapping[inIndex] = outIndex;
                        break;
                    }
                }
            }
            // Get the Index numbers of key columns in both the input and the output
            FindIndices();
        }
        #endregion

        #region ProcessInput
        /// <summary>
        /// This method is called repeatedly during package execution. It is called each time the data flow task
        /// has a full buffer provided by an upstream component. 
        /// </summary>
        /// <param name="inputID">The ID of the input object of the component.</param>
        /// <param name="buffer">The input buffer.</param>
        public override void ProcessInput(int inputID, PipelineBuffer buffer)
        { 
            IDTSInput100 input = ComponentMetaData.InputCollection.GetObjectByID(inputID);

            // This code sample assumes the component has two outputs, one the default,
            // the other the error output. If the errorOutputIndex returned from GetErrorOutputInfo
            // is 0, then the default output is the second output in the collection.
            int defaultOutputID = -1;
            int errorOutputID = -1;
            int errorOutputIndex = -1;
            GetErrorOutputInfo(ref errorOutputID, ref errorOutputIndex);

            if (errorOutputIndex == 0)
                defaultOutputID = ComponentMetaData.OutputCollection[1].ID;
            else
                defaultOutputID = ComponentMetaData.OutputCollection[0].ID;
            

            if (buffer == null)
            {
                throw new ArgumentNullException("buffer");
            }

            while (buffer.NextRow())
            {

                try
                {

                    //
                    // Check that this row matches the row currently in the OutputBuffer
                    //
                    if (!firstRow)
                    {
                        foreach (int i in keyColumn_Indices)
                        {
                            if (CompareColumns(outputBuffer[columnMapping[i]], buffer[i]) != 0)
                            {
                                rowMatches = false;
                            }
                        }
                    }
                    //
                    // if it is the first row or
                    // this row does not match the previous row's key
                    // columns, then create an output
                    //
                    if (firstRow == true || rowMatches == false)
                    {

                        // Copy data from input row to outputBuffer
                        // Add a row to the outputBuffer
                        outputBuffer.AddRow();

                        // Initialize aggregated text column to null ...
                        outputBuffer.SetNull(aggColumnOut_Index);

                        // ...then opy data from the nonaggregated column of the 
                        ///input buffer to the outputbuffer (all the mapped columns)
                        foreach (int inIndex in columnMapping.Keys)
                        {
                            object oValue = buffer[inIndex];
                            if (oValue != null)
                                if (!(oValue is BlobColumn))
                                {
                                    outputBuffer[columnMapping[inIndex]] = oValue;
                                }
                                // Blob (text) columns must be treated differently
                                else
                                {
                                    outputBuffer.AddBlobData(columnMapping[inIndex], enc.GetBytes(columnMapping[inIndex].ToString()));
                                }
                        }

                        // Now reset the rowMatches flag
                        rowMatches = true;

                    }


                    // if if there is text in the column we are aggregating
                    // add to the output!
                    // again, Blob data must be treated differently
                    if (!(buffer[aggColumnIn_Index] is BlobColumn))
                    {
                        if (buffer[aggColumnIn_Index] != null)
                        {
                            string stringToAdd = buffer[aggColumnIn_Index].ToString();
                            byte[] ByteArray = enc.GetBytes(stringToAdd);
                            outputBuffer.AddBlobData(aggColumnOut_Index, ByteArray);
                        }
                    }
                    else
                    {
                        BlobColumn InCol = (BlobColumn)buffer[aggColumnIn_Index];
                        outputBuffer.AddBlobData(aggColumnOut_Index, InCol.GetBlobData(0, (int)InCol.Length));
                    }

                    // Append Newline
                    outputBuffer.AddBlobData(aggColumnOut_Index, enc.GetBytes(newlinechar));



                    // since we have processed a row, set firstRow flag false
                    firstRow = false;
                }
                catch
                {


                            // Yes, has the user specified to redirect the row?
                            if (input.ErrorRowDisposition == DTSRowDisposition.RD_RedirectRow)
                            {
                                // Yes, direct the row to the error output.
                                // TODO: Add code to include the errorColumnIndex.
                                int errorColumnIndex = 0;
                                buffer.DirectErrorRow(errorOutputID, 100, errorColumnIndex);
                            }
                            else if (input.ErrorRowDisposition == DTSRowDisposition.RD_FailComponent || input.ErrorRowDisposition == DTSRowDisposition.RD_NotUsed)
                            {
                                // No, the user specified to fail the component, or the error row disposition was not set.
                                throw new Exception("An error occurred, and the DTSRowDisposition is either not set, or is set to fail component.");
                            }
                            else
                            {
                                // No, the user specified to ignore the failure so 
                                // direct the row to the default output.
                                buffer.DirectRow(defaultOutputID);
                            }

                        }

            }

            // EndOfRowset on the input buffer is true.
            // Set EndOfRowset on the output buffer.
            if (buffer.EndOfRowset)
            {
                outputBuffer.SetEndOfRowset();
            }
        }
        #endregion

        #region PrimeOutput
        /// <summary>
        /// Called after PreExecute. The data flow task provides a buffer for each output 
        /// that is connected to a downstream component. Check the location of each output specified in the
        /// outputIDs array, and based on the index of the output cache the output buffer to the appropriate
        /// private member variable.
        /// </summary>
        /// <param name="outputs">The number of outputs.</param>
        /// <param name="outputIDs">An array of output ids.</param>
        /// <param name="buffers">An array of output buffers.</param>
        public override void PrimeOutput(int outputs, int[] outputIDs, PipelineBuffer[] buffers)
        {
            if (buffers.Length != 0)
                outputBuffer = buffers[0];
            else
                throw new Exception("No output buffer defined.");

        }
        #endregion

        #endregion

        #region Helpers
        // little wrapper to fire an error message
        private void InternalFireError(string Message)
        {
            ComponentMetaData.FireError(0, ComponentMetaData.Name, Message, "", 0, out cancelEvent);
        }
        /// <summary>
        /// For each column in the input collection, add a corresponding column to the 
        /// output column collection of the specified output object.
        /// </summary>
        /// <param name="outputID">The ID of the output to add the columns to.</param>
        private void AddInputColumnsToOutput(int outputID)
        {
            IDTSInput100 input = ComponentMetaData.InputCollection[0];
            IDTSOutput100 output = ComponentMetaData.OutputCollection.GetObjectByID(outputID);

            output.OutputColumnCollection.RemoveAll();

            foreach (IDTSInputColumn100 Col in input.InputColumnCollection)
                if (Col.Name != GetCustomPropertyString("AggregateColumn"))
                    AddOutputColumn(outputID, Col);
        }



        /// <summary>
        /// Creates an output column in the output identified by the OutputID parameter.
        /// Adds a custom property to the output containing the lineageID of the input column. This property is
        /// used during Validate, to verify that each input column has a matching output column.
        /// </summary>
        /// <param name="OutputID">The ID of the output to create the column in.</param>
        /// <param name="inputColumn">The input column used to set the properties of the output column.</param>
        private void AddOutputColumn(int OutputID, IDTSInputColumn100 inputColumn)
        {
            IDTSOutput100 output = ComponentMetaData.OutputCollection.GetObjectByID(OutputID);
            IDTSOutputColumn100 outputColumn = output.OutputColumnCollection.New();
            outputColumn.Name = inputColumn.Name;

            // Store the lineageID of the input column in a custom property of the output column.
            IDTSCustomProperty100 inputColLineageID = outputColumn.CustomPropertyCollection.New();
            inputColLineageID.Name = "InputColumnLineageID";
            inputColLineageID.Value = inputColumn.LineageID;

            outputColumn.SetDataTypeProperties(inputColumn.DataType, inputColumn.Length, inputColumn.Precision, inputColumn.Scale, inputColumn.CodePage);
        }

        /// <summary>
        /// Removes the output column that references the input column. The reference is maintained in the 
        /// InputColumnLineageID custom property. 
        /// </summary>
        /// <param name="OutputID">The ID of the output.</param>
        /// <param name="inputColumn">The lineageID of the input column.</param>
        private void RemoveOutputColumn(int OutputID, int LineageID)
        {
            IDTSOutput100 output = ComponentMetaData.OutputCollection.GetObjectByID(OutputID);

            foreach (IDTSOutputColumn100 col in output.OutputColumnCollection)
            {
                if (CustomPropertyExists(col.CustomPropertyCollection, "InputColumnLineageID") &&
                    (int)col.CustomPropertyCollection["InputColumnLineageID"].Value == LineageID)
                {
                    output.OutputColumnCollection.RemoveObjectByID(col.ID);
                    break;
                }
            }



        }

        /// <summary>
        /// Searches the output column collection for a column containing the provided lineageID
        /// in the InputColumnLineageID custom property.
        /// </summary>
        /// <param name="LineageID">The lineageID of the input column.</param>
        /// <param name="output">The IDTSOutput to search.</param>
        /// <returns>true if found; otherwise false.</returns>
        private static bool OutputColumnExists(IDTSOutput100 output, int LineageID)
        {
            foreach (IDTSOutputColumn100 col in output.OutputColumnCollection)
            {

                if (CustomPropertyExists(col.CustomPropertyCollection, "InputColumnLineageID") &&
                    col.CustomPropertyCollection["InputColumnLineageID"].Value != null &&
                    (int)col.CustomPropertyCollection["InputColumnLineageID"].Value == LineageID)
                {
                    return true;
                }
            }

            return false;
        }

        private static int CompareColumns(object colA, object colB)
        {
            Type aType = colA.GetType();
            Type bType = colB.GetType();

            if (!aType.Equals(bType))
            {
                throw new Exception("Type mismatch when comparing columns in the internal buffer.");
            }

            if (aType.Equals(typeof(string)))
            {
                return ((string)colA).CompareTo((string)colB);
            }
            else if (aType.Equals(typeof(DateTime)))
            {
                return Convert.ToDateTime(colA, System.Globalization.CultureInfo.InvariantCulture).CompareTo(Convert.ToDateTime(colB, System.Globalization.CultureInfo.InvariantCulture));
            }
            else if (aType.Equals(typeof(decimal)))
            {
                return ((decimal)colA).CompareTo((decimal)colB);
            }
            else if (aType.Equals(typeof(Guid)))
            {
                return ((Guid)colA).CompareTo((Guid)colB);
            }
            else if (aType.Equals(typeof(Int16)))
            {
                return ((Int16)colA).CompareTo((Int16)colB);
            }
            else if (aType.Equals(typeof(Int32)))
            {
                return ((Int32)colA).CompareTo((Int32)colB);
            }
            else if (aType.Equals(typeof(Int64)))
            {
                return ((Int64)colA).CompareTo((Int64)colB);
            }
            else if (aType.Equals(typeof(bool)))
            {
                return ((bool)colA).CompareTo((bool)colB);
            }
            else if (aType.Equals(typeof(float)))
            {
                return ((float)colA).CompareTo((float)colB);
            }
            else if (aType.Equals(typeof(double)))
            {
                return ((double)colA).CompareTo((double)colB);
            }
            else if (aType.Equals(typeof(byte)))
            {
                return ((byte)colA).CompareTo((byte)colB);
            }
            else if (aType.Equals(typeof(UInt16)))
            {
                return ((UInt16)colA).CompareTo((UInt16)colB);
            }
            else if (aType.Equals(typeof(UInt32)))
            {
                return ((UInt32)colA).CompareTo((UInt32)colB);
            }
            else if (aType.Equals(typeof(UInt64)))
            {
                return ((UInt64)colA).CompareTo((UInt64)colB);
            }
            else
            {
                throw new Exception("Unsupported column data type.");
            }
        }

        private void FindIndices()
        {
            aggColumnIn_Index = -1;
            aggColumnOut_Index = -1;
            keyColumn_Indices = new List<int>();

            string aggColumnIn = ComponentMetaData.CustomPropertyCollection["AggregateColumn"].Value.ToString();
            string aggColumnOut = ComponentMetaData.CustomPropertyCollection["OutputColumn"].Value.ToString();


            //
            // Look at the input row and identify the index of important columns
            //  that will be used in this component
            //
            for (int i = 0; i < ComponentMetaData.InputCollection[0].InputColumnCollection.Count; i++)
            {
                int index = this.BufferManager.FindColumnByLineageID(ComponentMetaData.InputCollection[0].Buffer, ComponentMetaData.InputCollection[0].InputColumnCollection[i].LineageID);
                string name = ComponentMetaData.InputCollection[0].InputColumnCollection[i].Name.ToLower();

                // Identify the index of the column containing text 
                // to aggregate
                if (name == aggColumnIn.ToLower())
                    aggColumnIn_Index = index;

                // Indentify the index of each of they key columns 
                // that we group by
                foreach (string keyCol in GetCustomPropertyString("keyColumns").ToLower().Split(','))
                {
                    if (name == keyCol.ToLower())
                    {
                        keyColumn_Indices.Add(index);
                    }

                }


            }

            //
            // Identify the index of the output column containing the aggregated text
            //
            for (int i = 0; i < ComponentMetaData.OutputCollection[0].OutputColumnCollection.Count; i++)
            {
                int index = this.BufferManager.FindColumnByLineageID(ComponentMetaData.OutputCollection[0].Buffer, ComponentMetaData.OutputCollection[0].OutputColumnCollection[i].LineageID);

                string name = ComponentMetaData.OutputCollection[0].OutputColumnCollection[i].Name.ToLower();

                if (name == aggColumnOut.ToLower())
                    aggColumnOut_Index = index;
            }


        }

        /// <summary>
        /// Searches the output column collection for a column containing the provided lineageID
        /// in the InputColumnLineageID custom property.
        /// </summary>
        /// <param name="CustomPropertyCollection">The custom property collection to search.</param>
        /// <param name="CustomPropertyName">The name of the custom property to search for.</param>
        /// <returns>true if found; otherwise false.</returns>
        private static bool CustomPropertyExists(IDTSCustomPropertyCollection100 CustomPropertyCollection, string CustomPropertyName)
        {
            foreach (IDTSCustomProperty100 property in CustomPropertyCollection)
            {
                if (property.Name == CustomPropertyName)
                {
                    return true;
                }
            }

            return false;
        }

        private void AddTextOutputColumn()
        {
            if (GetCustomPropertyString("OutputColumn") != null)
            {

                // add text column output
                IDTSOutputColumn100 output_txt = ComponentMetaData.OutputCollection[0].OutputColumnCollection.New();
                output_txt.Name = GetCustomPropertyString("OutputColumn");
                output_txt.SetDataTypeProperties(DataType.DT_TEXT, 0, 0, 0, 1252);

                // Store the lineageID of the input column in a custom property of the output column.
                IDTSCustomProperty100 inputColLineageID = output_txt.CustomPropertyCollection.New();
                inputColLineageID.Name = "InputColumnLineageID";
                inputColLineageID.Value = null;
            }
        }

        /// <summary>
        /// Removes the output column that has the specified Name
        /// </summary>
        /// <param name="Output">The output to search.</param> 
        /// <param name="NameToRemove">The name of the column to revmoe</param>

        /// 
        private void RemoveOutputColumnByName(IDTSOutput100 Output, string NameToRemove)
        {

            foreach (IDTSOutputColumn100 col in Output.OutputColumnCollection)
            {
                if (col.Name == NameToRemove)
                {
                    Output.OutputColumnCollection.RemoveObjectByID(col.ID);
                    break;
                }
            }



        }

        /// <summary>
        /// Searches the output column collection for a column with the name of
        /// a given string
        /// </summary>
        /// <param name="Output">The output to search.</param>
        /// <param name="ColumnName">The name of the column to search for.</param>
        /// <returns>true if found; otherwise false.</returns>
        private static bool OutputColumnExistsByName(IDTSOutput100 Output, string ColumnName)
        {
            foreach (IDTSOutputColumn100 col in Output.OutputColumnCollection)
            {
                if (col.Name == ColumnName)
                {
                    return true;
                }
            }

            return false;
        }

        /// <summary>
        /// Searches the output column collection for a column with the name of
        /// a given string
        /// </summary>
        /// <param name="Input">The Input to search.</param>
        /// <param name="ColumnName">The name of the column to search for.</param>
        /// <returns>true if found; otherwise false.</returns>
        private static bool InputColumnExistsByName(IDTSInput100 Input, string ColumnName)
        {
            foreach (IDTSInputColumn100 col in Input.InputColumnCollection)
            {
                if (col.Name == ColumnName)
                {
                    return true;
                }
            }

            return false;
        }


        /// <summary>
        /// Returns a string value of a custom property.  Returns null if the custom property 
        /// does not exists or the value is not set
        /// </summary>
        /// <param name="CustomProperty">The CustomProperty to return.</param>
        /// <returns>true if found; otherwise false.</returns>
        private string GetCustomPropertyString(string CustomProperty)
        {
            if (ComponentMetaData.CustomPropertyCollection[CustomProperty] == null)
                return null;

            else if (ComponentMetaData.CustomPropertyCollection[CustomProperty].Value == null)
                return null;
            else
                return ComponentMetaData.CustomPropertyCollection[CustomProperty].Value.ToString();

        }


        #endregion


    }
}